/*
    Copyright (C) 2021 Huawei Device Co., Ltd.
    Licensed under the Apache License, Version 2.0 (the "License");
    you may not use this file except in compliance with the License.
    You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

    Unless required by applicable law or agreed to in writing, software
    distributed under the License is distributed on an "AS IS" BASIS,
    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    See the License for the specific language governing permissions and
    limitations under the License.
 */

package com.danikula.videocache.slicingStrategy;

import com.danikula.videocache.*;
import com.danikula.videocache.util.ProxyCacheUtils;

import java.io.BufferedOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.net.Socket;
import java.util.ArrayList;
import java.util.List;

/**
 * 文件分片缓存代理
 */
public class HttpSliceProxyCache extends ProxyCache {
    private final HttpUrlSource source;
    private final SlicingFileCache cache;

    public HttpSliceProxyCache(HttpUrlSource source, SlicingFileCache cache) {
        super(source, cache);
        this.cache = cache;
        this.source = source;
    }

    @Override
    public void processRequest(GetRequest request, Socket socket) throws IOException, ProxyCacheException,
            InterruptedException {
        OutputStream out = new BufferedOutputStream(socket.getOutputStream());
        String responseHeaders = newResponseHeaders(request);
        out.write(responseHeaders.getBytes("UTF-8"));
        long offset = request.rangeOffset;
        System.out.println("测试打印===offset   "+offset );
        if (cache.isTempFile(cache.file)) {
            if (source.length() > 0) {
                System.out.println("测试打印===FileLength   "+source.length() );
                cache.setFileLength(source.length());
            }
            cache.setCacheOffset(offset);
            responseWithCache(out, offset);
        } else {
            onCachePercentsAvailableChanged(100);
            responseWithFile(out, offset);
        }
    }

    @Override
    public int read(byte[] buffer, long offset, int length) throws ProxyCacheException {
        ProxyCacheUtils.assertBuffer(buffer, offset, length);
        while (!cache.isCompleted() && !cache.isCacheEnough(offset, length) && !stopped) {
            readSourceAsync();
            waitForSourceData();
            checkReadSourceErrorsCount();
        }
        int read = cache.readFile(buffer, offset, length);
        if (cache.isCompleted() && percentsAvailable != 100) {
            percentsAvailable = 100;
            onCachePercentsAvailableChanged(100);
        }
        return read;
    }


    @Override
    protected void onCachePercentsAvailableChanged(int percents) {
        long fileTotalLength = 0;
        List<long[]> allCacheArea = new ArrayList<>();
        if (cache != null) {
            fileTotalLength = cache.getFileLength();
            allCacheArea = cache.getAllCacheArea();
        }
        if (listener != null && fileTotalLength > 0 && !allCacheArea.isEmpty()) {
            listener.onCachePartAvailable(cache.file, source.getUrl(), allCacheArea,
                    fileTotalLength);
        }
    }

    @Override
    protected Runnable getRunable() {
        return new SlicingSourceRunnable();
    }

    private class SlicingSourceRunnable implements Runnable {
        long sourceAvailable = -1;
        long offset = 0;
        long mergeStart = 0;

        @Override
        public void run() {
            readSource();
        }

        private void readSource() {
            try {
                if (!cache.isMergeFinish()) {
                    onError(new ProxyCacheException("文件合并未结束"));
                    return;
                }
                offset = cache.getDownLoadStart();
                if (offset >= cache.getFileLength() && offset > 0) {
                    onError(new ProxyCacheException("下载完毕"));
                    return;
                }
                source.open(offset);
                sourceAvailable = source.length();
                cache.setFileLength(sourceAvailable);
                if (offset == sourceAvailable) {
                    onError(new ProxyCacheException("当前碎片下载完毕"));
                    return;
                }
                cache.refreshMergeInfo(offset);
                writeData();
                tryComplete();
                onSourceRead();
            } catch (ProxyCacheException e) {
                readSourceErrorsCount.incrementAndGet();
                onError(e);
            } finally {
                if (cache != null) {
                    cache.setMergeFinish(true);
                }
                closeSource();
                notifyNewCacheDataAvailable(offset, sourceAvailable);
            }
        }

        private void writeData() throws ProxyCacheException {
            byte[] buffer = new byte[ProxyCacheUtils.DEFAULT_BUFFER_SIZE];
            int readBytes;
            mergeStart = cache.getMergeStart();
            while ((readBytes = source.read(buffer)) != -1) {
                synchronized (stopLock) {
                    if (isStopped()) {
                        return;
                    }
                    cache.writeFile(buffer, offset, readBytes);
                    offset += readBytes;
                    /**
                     * offset >= mergeStart
                     * 当本缓存碎片下载量达到距离它最近的可以合并的文件的起点的时候 停止下载
                     * 在writeFile之后才return 允许超过拼接点写入数据 合并的时候会覆盖
                     */
                    if (offset >= mergeStart) {
                        break;
                    }
                    if (offset >= sourceAvailable) {
                        break;
                    }
                }
                notifyNewCacheDataAvailable(cache.getTotalCache(), sourceAvailable);
            }
        }
    }


    @Override
    protected void tryComplete() throws ProxyCacheException {
        synchronized (stopLock) {
            stopped = true;
            cache.mergeFiles();
            if (cache.getTotalCache() == source.length()) {
                cache.complete();
            }
        }
    }

    @Override
    public void doComplete() throws ProxyCacheException {
        tryComplete();
        onSourceRead();
    }

    @Override
    public void onSourceRead() {
        percentsAvailable = (int) (100 * cache.getTotalCache() / cache.getFileLength());
        onCachePercentsAvailableChanged(percentsAvailable);
    }

}
